package net.bwie.flink;

import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Collections;
import java.util.concurrent.*;
import java.util.function.Supplier;

/**
 *
 * @author xuanyu
 * @date 2025/10/29
 */
public class AsyncDatabaseRequest extends RichAsyncFunction<String, String> {

	// 创建线程池
	private transient ExecutorService executorService = Executors.newFixedThreadPool(10) ;

	@Override
	public void asyncInvoke(String input, ResultFuture<String> resultFuture) throws Exception {
		// 1. 获取skuName  input -> u101,o1,milk,5.0,2025-10-29 17:30:10
		String skuName = input.split(",")[2];
		// 2. 线程池请求
		Future<String> future = executorService.submit(
			new Callable<String>() {
				@Override
				public String call() throws Exception {
					// 返回
					return requestMysql(skuName);
				}
			}
		);
		// 3.返回
		CompletableFuture.supplyAsync(
			new Supplier<String>() {
				@Override
				public String get() {
					try {
						return future.get();
					} catch (Exception e) {
						throw new RuntimeException(e);
					}
				}
			}
		).thenAccept(
			result -> resultFuture.complete(Collections.singleton(result))
		);
	}

	@Override
	public void timeout(String input, ResultFuture<String> resultFuture) throws Exception {
		super.timeout(input, resultFuture);
	}

	/**
	 * 请求MySQL，依据skuName获取categoryName名称
	 *      SELECT category_name FROM flink_day13.sku_info WHERE sku_name = ? ;
	 */
	private static String requestMysql(String skuName) throws Exception {
		// s0.变量
		Connection conn = null;
		PreparedStatement pstmt = null ;
		ResultSet resultSet = null;
		String categoryName = "unknown" ;
		try {
			// s1.连接
			conn = DriverManager.getConnection(
				"jdbc:mysql://node101:3306/flink_day13", "root", "123456"
			);
			// s2.语句
			pstmt = conn.prepareStatement(
				"SELECT category_name FROM flink_day13.sku_info WHERE sku_name = ?"
			);
			pstmt.setString(1, skuName);
			// s3-执行
			resultSet = pstmt.executeQuery();
			// s4-获取
			while (resultSet.next()){
				categoryName = resultSet.getString("category_name");
			}
		}catch (Exception e){
			e.printStackTrace();
		}finally {
			// s5.关闭
			resultSet.close();
			pstmt.close();
			conn.close();
		}
		return categoryName;
	}

}
